import sys,os,time
import marshal

from threading import Thread, Lock
from Queue import Queue

class AtomicCounter(object):
    def __init__(self, value = 0):
        self.__value = value
        self.lock = Lock()
    
    '''
        atomic add
    '''
    def incr(self, n = 1):
        with self.lock:
            self.__value += n
    
    '''
        atomic decrease
    '''
    def decr(self, n = 1):
        self.incr(-1)
        
    def value(self):
        return self.__value
        
class JobState(object):
    def __init__(self, f, fid = None):
        assert(f)
        self.f = f
        self.fid = fid

    def __call__(self):
        
        if self.fid is None:
            self.fid = __file__ + '.' + self.f.func_name + '.dump'
            
        result = None
        try:
            f = open(self.fid, 'rb')
            result = marshal.load(f)
            f.close()
        except:
            pass
        
        if result is None:
            result = self.f()
            if result:
                try:
                    f = open(self.fid, 'wb')
                    marshal.dump(result, f)
                except:
                    pass
        
        return result

class JobRunner(object):
    def __init__(self, producer, consumer, consumersize = 4):
        assert(producer)
        assert(consumer)
        assert(consumersize > 0)
        self.producer = producer
        self.consumer = consumer
        self.consumersize = consumersize
        
        self.q = Queue()
        self.totaljobs = AtomicCounter(0)
        self.finishedjobs = AtomicCounter(0)
        self.threads = []
        self.running = False
        
    def run(self):
        assert(not self.running)
        
        self.running = True
        self.t0 = time.time()
        
        def _p(): # producer thread
            for j in self.producer():
                self.q.put(j)
                self.totaljobs.incr()
        
        def _c(): # consumer thread
            while True:
                j = self.q.get()
                self.consumer(j)
                self.finishedjobs.incr()
                self.q.task_done()
                
        def _stat(): # state
            while True:
                time.sleep(1)
                total = self.totaljobs.value()
                finished = self.finishedjobs.value()
                if total > 0:
                    print '%7.1fs - jobs(done/total)=%d/%d(%.2f)%%' % (
                        time.time() - self.t0, finished, total, finished * 100.0 / total)
        
        workers = [_p, _stat]
        workers.extend([_c] * self.consumersize)

        # start all the workers
        for w in workers:
            t = Thread(target=w)
            t.daemon = True
            t.start()
            self.threads.append(t)
        
    def join(self):
        self.q.join()
        print 'jobs done in %.2fs' % (time.time() - self.t0)

if __name__ == '__main__': # test
    def p():
        return xrange(1067)
    l = Lock()
    def c(i):
        s = (i % 5) / 500.0
        time.sleep(s)
        with l:
            print i
    jr = JobRunner(p,c)
    jr.run()
    jr.join()